
package test;

import cfar.actors.Message;
import cfar.actors.Node;
import cfar.actors.distributed.ActorServer;
import cfar.ff.data.Assocs.Assoc2;
import cfar.ff.data.List;
import cfar.ff.fun.F0;
import cfar.ff.fun.F1;
import cfar.ff.fun.F2;
import cfar.ff.fun.Seq;
import cfar.ff.interfaces.Sequence;
import cfar.utils.IO;

import static cfar.ff.fun.Curry.*;

import java.io.IOException;
import java.net.InetAddress;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 * @author badi
 */
public class Main {

    public static F1 mapper = new F1<Integer, Integer>() {

        public Integer f(Integer a) {
            return a + 10;
        }
    };

    public static F2<Integer,Integer,Integer> reducer = new F2<Integer, Integer, Integer>() {

        public Integer f(Integer a, Integer b) {
            return a + b;
        }
    };

    public static F2<String,String,Sequence<List,Node>> get_nodes = new F2<String, String, Sequence<List,Node>>() {

        public Sequence<List,Node> f(String dir, String suffix) {
            Sequence<List,Assoc2<InetAddress,Integer>> locs = IO.available_hosts(syncdir, suffix);
            List<Node> ns = List.nil();
            for(Assoc2<InetAddress,Integer> l : locs){
                try {
                    ns = ns.cons(new Node(l.$1, l.$2));
                } catch (IOException ex) {
                    Logger.getLogger(Main.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
            return ns;
        }
    };

    static String syncdir = "/tmp/cfar.sync"; // "/dscratch/cabdulwa/os/cfar.sync";
    static String suffix  = "cfar";

    /**
     * @param args the command line arguments
     */
    public static void main(String[] args) throws IOException {
        if(args.length == 0)
            for (int i = 0; i < 10; i++)
                new Thread(new ActorServer(syncdir, suffix)).start();
        else {
            List l = List.list(1,2,3,4,5,6,7,8,9,10);

            Sequence<List,Assoc2<InetAddress, Integer>> locs = IO.available_hosts(syncdir, suffix);
            List<Node> nodes = List.nil();
            for(Assoc2 a : locs)
                nodes = nodes.cons(new Node((InetAddress) a.$1, (Integer) a.$2));

            System.out.println(Seq.distMap(nodes, mapper, l));

            F0 getnodes = curry(curry(get_nodes, syncdir), suffix);
            System.out.println(Seq.distReduce(getnodes, reducer, l));

            for(Node n : nodes)
                n.tell(Message.stop);
        }
    }

}
